package com.spark.mlib.gb

import java.util

import com.mongodb.{BasicDBObject, DBCollection, DBObject}
import com.mongodb.casbah.MongoClient
import org.apache.log4j.{Level, Logger}
import org.apache.spark.ml.feature.{IndexToString, OneHotEncoder, StringIndexer}
import org.apache.spark.mllib.tree.GradientBoostedTrees
import org.apache.spark.mllib.tree.configuration.BoostingStrategy
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.sql.{SQLContext, SparkSession}
import redis.clients.jedis.{Jedis, JedisPoolConfig}
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.regression.LabeledPoint

import scala.collection.{JavaConversions, mutable}

object GBTdemo1 {
  def getDayPhase(dayPhase: String): Int = dayPhase match {
    case "am" => 1;
    case "noon" => 2;
    case "pm" => 3;
    case "night" => 4;
    case "deepNight" => 5;
    case _ => 0;
  }

  def getDayType(dayType: String): Int = dayType match {
    case "workday" => 1;
    case "weekend" => 2;
    case _ => 0;
  }

  def getPlatform(platform: String): Int = platform match {
    case "ios" => 1;
    case "android" => 2;
    case _ => 0;
  }

  def getDeviceNetwork(deviceNetwork: String): Int = deviceNetwork match {
    case "3g" => 1;
    case "4g" => 2;
    case "wifi" => 3;
    case _ => 0;
  }

  def getPayUser(pay: String): Int = pay match {
    case "true" => 1;
    case _ => 0;
  }

  def getConsumeTimes(times: Int): Int = times match {
    case _ if times == 1 => 1
    case _ if times > 1 => 2
    case _ => 0
  }

  def getJedis(): Jedis = {
    val config: JedisPoolConfig = new JedisPoolConfig
    config.setMaxIdle(10)
    import redis.clients.jedis.JedisPool
    var pool = new JedisPool(config, "localhost", 6379)
    pool.getResource()
  }

  def getDBField(): BasicDBObject = {
    val dBObject = new BasicDBObject()
    //deviceID,appID,appType,deviceBrand,country,province,city,platform,day_phase,day_type,device_network,adID,ad_scene,pay_user,consume_times
    dBObject.append("deviceID", 1)
    dBObject.append("appID", 1)
    //dBObject.append("appType", 1)
    //dBObject.append("deviceBrand", 1)
    //dBObject.append("country", 1)
    //dBObject.append("province", 1)
    //dBObject.append("city", 1)
    dBObject.append("platform", 1)
    dBObject.append("day_phase", 1)
    dBObject.append("day_type", 1)
    dBObject.append("device_network", 1)
    //dBObject.append("adID", 1)
    //dBObject.append("ad_scene", 1)
    dBObject.append("pay_user", 1)
    dBObject.append("consume_times", 1)
    dBObject
  }

  def getKeyIndex(): mutable.HashMap[String, Int] = {
    val keyMap = new mutable.HashMap[String, Int]()
    keyMap.put("deviceID", 1)
    //keyMap.put("appID", 2)
    //keyMap.put("appType", 3)
    //keyMap.put("deviceBrand", 4)
    //keyMap.put("country", 5)
    //keyMap.put("province", 6)
    //keyMap.put("city", 7)
    keyMap.put("platform", 2)
    keyMap.put("day_phase", 3)
    keyMap.put("day_type", 4)
    keyMap.put("device_network", 5)
    //keyMap.put("adID", 12)
    //keyMap.put("ad_scene", 13)
    keyMap.put("pay_user", 6)
    keyMap.put("consume_times", 7)
    keyMap
  }
  def getAppType(c :  DBCollection, appId : String) : String = {
    val condition = new BasicDBObject()
    condition.append("appID",appId)
    val field = new BasicDBObject()
    field.append("appID",1)
    field.append("appType",1)
    val data = c.find(condition,field)
    if(data.size() > 0){
      data.next().get("appType").toString
    }
    else{
      "unknown"
    }
  }
  case class MyFeature(did: String, platform: Double, dayPhase: Double, dayType: Double, dNetwork: Double, payUser: Double, consumeTimes: Double)

  def main(args: Array[String]): Unit = {
    // deviceID,appID,appType,deviceBrand,country,province,city,platform,
    // day_phase,day_type,device_network,adID,ad_scene,pay_user,consume_times
    //    label： 愿意付费-1，流量多-2,夜猫子-3，游戏爱好者-4，自由-5，买买买-6
    val mongoClient = MongoClient()
    val mongoDatabase = mongoClient.getDB("testdb")
    val collection = mongoDatabase.getCollection("daily_user_info")
    println(collection.find().count())
    val condition = new BasicDBObject()
    condition.append("appID",new BasicDBObject("$ne",null))
    val dBObject = getDBField()
    val keyMap = getKeyIndex()
    //println(collection.find(condition, dBObject).one())
    val appCollection = mongoDatabase.getCollection("app_info")

    val sourceData = collection.find(condition, dBObject)
    val listFeature = new util.ArrayList[MyFeature]()

    val set = scala.collection.mutable.Set("1")

    while (sourceData.hasNext) {
      val row = sourceData.next().toMap
      val did = row.get("deviceID").toString
      var platform = 0
      if (row.containsKey("platform") && row.get("platform") != null) {
        platform = getPlatform(row.get("platform").toString)
      }
      var dayPhase = 0
      if (row.containsKey("day_phase") && row.get("day_phase") != null) {
        dayPhase = getDayPhase(row.get("day_phase").toString)
      }
      var dayType = 0
      if (row.containsKey("day_type") && row.get("day_type") != null) {
        dayType = getDayType(row.get("day_type").toString)
      }
      var dnetwork = 0
      if (row.containsKey("dnetwork") && row.get("dnetwork") != null) {
        dnetwork = getDeviceNetwork(row.get("device_network").toString)
      }

      var payUser = 0
      if (row.containsKey("pay_user") && row.get("pay_user") != null) {
        payUser =  getPayUser(row.get("pay_user").toString)
      }
      var consumeTimes = 0
      if (row.containsKey("consume_times") && row.get("consume_times") != null) {
        consumeTimes = getConsumeTimes(row.get("consume_times").toString.toInt)
      }
      println(row.toString)
      val appID = row.get("appID").toString
      val appType = getAppType(appCollection,appID)
      set.add(appType)
      println(appID+"="+appType)
      println(set.size)

      listFeature.add(MyFeature(did, platform, dayPhase, dayType, dnetwork, payUser, consumeTimes))

    }
    var a = 0;
    while (a < 10) {
      println(listFeature.get(a))
      a += 1
    }
    // train model by sample data
    val conf = new SparkConf().setMaster("local").setAppName("gbt1")
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)
    val data = MLUtils.loadLibSVMFile(sc, this.getClass.getClassLoader.getResource("libsvm2.txt").getFile)
    //Classification 分类，Regression 回归
    val gbType = "Regression"
    val boostingStrategy = BoostingStrategy.defaultParams(gbType)
    boostingStrategy.numIterations = 5
    //boostingStrategy.treeStrategy.numClasses = 2//if Regression comment
    boostingStrategy.treeStrategy.maxDepth = 4
    boostingStrategy.treeStrategy.categoricalFeaturesInfo = Map[Int, Int]()
    //val model = GradientBoostedTrees.train(data,boostingStrategy)
    //println("Learned "+gbType+" GBT model:\n" + model.toDebugString)

    val df = sc.parallelize(JavaConversions.asScalaBuffer(listFeature))
    val sparkSession = SparkSession.builder().appName("RDD to DataFrame").config(conf).getOrCreate()
    import sparkSession.implicits._
    val dataFrame = df.toDF()
    val feartureIndex = List("platform","dayPhase","dayType","dNetwork","payUser","consumeTimes").map(dataFrame.columns.indexOf(_))

    val labelIndex = dataFrame.columns.indexOf("did")
    val lps = dataFrame.rdd.map(r => LabeledPoint(
      1,//default is 1， r.getDouble(labelIndex), // Get label value
      Vectors.dense(feartureIndex.map(r.getDouble(_)).toArray)
    ))
    //lps.foreach(println)
    //lps.foreach(x => println(model.predict(x.features)+" "+x.features))

    //    val indexer = new StringIndexer().setInputCol("dayPhase").setOutputCol("dayPhaseIndex")
    //    val model = indexer.fit(dataFrame)
    //    val indexerDayPhase = model.transform(dataFrame)
    //    indexerDayPhase.show()
    //数值化
    //    val converter = new IndexToString().setInputCol("dayPhaseIndex").setOutputCol("originaldayPhase")
    //    //反数值化
    //    val converted = converter.transform(indexerDayPhase)
    //    converted.show()
    //one-hot
    //    val encoder = new OneHotEncoder().setInputCol("dayPhaseIndex").setOutputCol("dayPhaseVec")
    //    val encoded = encoder.transform(indexerDayPhase)
    //    encoded.show()
  }
}
